package com.yesun.task.pipeline.sink;

import com.google.common.collect.Queues;
import com.yesun.task.pipeline.domain.dto.TaskDto;
import com.yesun.task.pipeline.domain.po.DetailPO;
import com.yesun.task.pipeline.domain.po.NewsPO;
import com.yesun.task.pipeline.service.impl.ClickhouseServiceImpl;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Author yesun
 * @Descritpion
 * @Date 2023/8/25 21:52
 * @Version 1.0
 **/
@Slf4j
public class ClickhouseSink {

    //交易明细
    public static LinkedBlockingQueue<String> QUEUE = new LinkedBlockingQueue();
    //新闻
    public static LinkedBlockingQueue<NewsPO> QUEUE_NEWS = new LinkedBlockingQueue();
    private static volatile boolean isRunning = true;
    //启动一个线程批量写出到Clickhouse
    public static class BatchInsertWorker extends Thread {
        @Override
        public void run() {
            log.info("BatchInsertWorker staring...");
            ClickhouseServiceImpl clickhouseServiceImpl = (ClickhouseServiceImpl) com.yesun.task.pipeline.common.SpringUtil.APPLICATION_CONTEXT.getBean("clickhouseServiceImpl");
            int i = 0;
            long last = System.currentTimeMillis();
            while (isRunning) {
                try {
                    List<String> listMsg = new ArrayList();
                    //每次到1000条数据才进行入库，等待1分钟，没达到1000条也继续入库
                    Queues.drain(QUEUE, listMsg, 1000, 1, TimeUnit.MINUTES);
                    List<DetailPO> list = new ArrayList();
                    for(String msg : listMsg) {
                        String[] arr = msg.split(",");
                        DetailPO detailPO = new DetailPO();
                        detailPO.setTradeDate(arr[0]);
                        detailPO.setDt(arr[0]);
                        detailPO.setTsCode(arr[1]);
                        detailPO.setTradeDateTime(arr[2]);
                        detailPO.setPrice(Float.parseFloat(arr[3]));
                        detailPO.setVol(arr[4]);
                        detailPO.setTradeCount(Integer.parseInt(arr[5]));
                        detailPO.setTradeType(Integer.parseInt(arr[6]));
                        list.add(detailPO);
                    }
                    if(list.size()>0) {
                        //插入
                        log.info("batch insert to clickhouse!!! msg size:" + list.size());
                        clickhouseServiceImpl.batchInsert(list);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }



    //启动一个线程批量写出到Clickhouse
    public static class BatchInsertNewsWorker extends Thread {
        @Override
        public void run() {
            log.info("BatchInsertNewsWorker staring...");
            ClickhouseServiceImpl clickhouseServiceImpl = (ClickhouseServiceImpl) com.yesun.task.pipeline.common.SpringUtil.APPLICATION_CONTEXT.getBean("clickhouseServiceImpl");
            int i = 0;
            long last = System.currentTimeMillis();
            while (isRunning) {
                try {
                    List<NewsPO> list = new ArrayList();
                    //每次到1000条数据才进行入库，等待1分钟，没达到1000条也继续入库
                    Queues.drain(QUEUE_NEWS, list, 10, 1, TimeUnit.MINUTES);
                    if(list.size()>0) {
                        //插入
                        log.info("batch insert news to clickhouse!!! size:" + list.size());
                        clickhouseServiceImpl.batchInsertNews(list);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}
